package com.mjf.day4

import com.mjf.day3.SensorReading
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * ProcessFunction:
 *  1.唯一可以获取到时间相关信息的API
 *  2.RichFunction能做的ProcessFunction都能做
 *  3.另外，可以获取到timestamp和watermark
 *  4.可以注册定时器，指定某个时间点发生的操作
 *  5.还可以输出侧输出流
 */
object ProcessFunctionExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream: DataStream[SensorReading] =
      env.socketTextStream("hadoop103", 9999)
        .map {
          line =>
            val strings: Array[String] = line.split(",")
            SensorReading(strings(0), strings(1).toLong, strings(2).toDouble)
        }

    // 检测每一个传感器温度在10秒之内没有下降时报警
    val resultDataStream: DataStream[String] = inputStream
      .keyBy("id") // 写"id"时，下面KeyedProcessFunction的泛型需要写Tuple
//        .keyBy(_.id)  // 写_.id时，下面KeyedProcessFunction的泛型需要写String
      .process(new TempIncreWarning(10000L))


    resultDataStream.print()

    env.execute("ProcessFunctionExample")

  }
}

// 自定义KeyedProcessFunction
class TempIncreWarning(interval: Long) extends KeyedProcessFunction[Tuple, SensorReading, String] {
  // 由于需要跟之前的温度做对比，所以将上一个温度保存成状态
  lazy val lastTempStat: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))
  // 为了方便删除定时器，还需要保存定时器的时间戳
  lazy val curTimerTsState: ValueState[Long] = getRuntimeContext.getState(new ValueStateDescriptor[Long]("cur-timer-ts", classOf[Long]))

  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[Tuple, SensorReading, String]#Context, out: Collector[String]): Unit = {
    // 首先取出状态
    val lastTemp = lastTempStat.value()
    val curTimerTs = curTimerTsState.value()

    // 将上次温度的状态更新为当前数据的温度值
    lastTempStat.update(value.temperature)

    // 判断当前温度值，如果比之前温度高，并且没有定时器的话，注册10秒后的定时器
    if (value.temperature > lastTemp && curTimerTs == 0) {
      val ts: Long = ctx.timerService().currentProcessingTime() + interval
      ctx.timerService().registerProcessingTimeTimer(ts) // 注册定时器
      curTimerTsState.update(ts) // 更新状态值
    } else if (value.temperature < lastTemp) { // 如果温度下降
      ctx.timerService().deleteProcessingTimeTimer(curTimerTs) // 删除定时器
      curTimerTsState.clear() // 清空状态
    }
  }

  // 定时器触发，说明10秒内没有下降的温度值，报警
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Tuple, SensorReading, String]#OnTimerContext, out: Collector[String]): Unit = {
    out.collect(ctx.getCurrentKey + "温度值连续" + interval / 1000 + "秒没有下降")
    curTimerTsState.clear() // 清空状态
  }
}